package org.wejob.core.log;

import org.wejob.core.log.util.BufferPool;
import org.wejob.core.log.util.SystemTime;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.LinkedList;
import java.util.concurrent.locks.ReentrantLock;

public class Writer extends Thread {
    ConcurrentList<Topic> tasks;
    int handled;
    BufferPool bufferPool;
    LinkedList<ByteBuffer>toWrite;
    Mqueue mqueue;

    public Writer() {
        //todo;内存池设计
        bufferPool=new BufferPool(32*1024*1024,1,new SystemTime(),"pool");
    }
    public Writer(ConcurrentList topics,Mqueue mqueue){
        this();
        this.tasks=topics;
        this.mqueue=mqueue;
    }

    @Override
    public void run() {
        for(;;){
            try {
                if(runOnce()<1){
                    synchronized (this){System.out.println("writer "+this.getName()+"  wait");
                        this.mqueue.waiters.put(this);
                        this.wait();

                    }
                }else{
                    this.handled=0;
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public int taskNum(){
        return this.tasks.size;
    }

    //这里必须保持同步
    public void acceptTask(Topic topic){
        this.tasks.put(topic);
    }


    public int runOnce() throws InterruptedException {
        int size=this.tasks.size;
        //handled计算当待处理的topic的个数
        this.tasks.forEach(topic -> {
            try {
                int var=topic.writeAndCode();
                if(var>-1){
                    //队列为空
                    return;
                };

                if(var<0){
                    //队列不为空,内存写满了;本次处理完毕
                    this.handled++;
                }
            } catch (IOException | InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        return handled;
    }
    private void writeMessage(){

    }
    public void doWrite(MessageRecord record){

    }
    private int doWrite(Topic topic) throws IOException, InterruptedException {
        //每次贪心处理topic中的消息,直到写满一个内存页或者队列为空
        int result=0;
        for(;;){
            MessageRecord record=topic.messageQueue.poll();
            if(record==null){
                topic.releaseCodeLock();
                break;
            }
            record.offset=topic.offsetCounter();
            if(topic.memoryPage ==null){
                topic.memoryPage =bufferPool.allocate(1024,1000);
            }
            byte[] serializeMessage=serializeMessage(record);
            int size=serializeMessage.length;
            //当前缓存页足够时直接写入
            if(serializeMessage.length<=topic.memoryPage.limit()-topic.memoryPage.position()){
                result++;
                topic.memoryPage.put(serializeMessage);
                topic.maxCommitedOffset= record.offset;
            }else if(size> this.bufferPool.poolableSize()){//size 大于pagesize
                ByteBuffer old=topic.memoryPage;
                MemoryRecords batch=new MemoryRecords(old, topic.maxCodedOffset);
                topic.toWriteTasks.offer(batch);
                topic.memoryPage =null;
                ByteBuffer byteBuffer=this.bufferPool.allocate(size,1000);
                byteBuffer.put(serializeMessage);
                MemoryRecords batch2=new MemoryRecords(byteBuffer,record.offset);
                topic.toWriteTasks.offer(batch2);
                if(topic.tryFileLock()){
                    topic.releaseCodeLock();
                    ByteBuffer var1=topic.toWriteTasks.poll().buffer;
                    while(var1!=null){
                        topic.fileChannel.write(var1);
                        bufferPool.deallocate(var1);
                        var1=topic.toWriteTasks.poll().buffer;
                    }
                    topic.releaseFileLock();
                }
                return -1;
            }else{
                ByteBuffer old=topic.memoryPage;
                topic.memoryPage =this.bufferPool.allocate(size,1000);
                topic.memoryPage.put(serializeMessage);
                if(topic.tryFileLock()){
                    topic.releaseCodeLock();
                    topic.fileChannel.write(old);
                    bufferPool.deallocate(old);
                    topic.releaseFileLock();
                }
                return -1;
            }
           /* //回调相关---暂不处理
            record.commited=true;
            //调用callback--ToDo--这里考虑将callback放到线程池中执行
            if(record.callback!=null){
                topic.queue.force();
                record.callback.onCompletion("success",null);
            };*/
        }
        return result;
    }
    private byte[] serializeMessage(MessageRecord message){
        //message.
        return message.message().toString().getBytes(StandardCharsets.UTF_8);
    }

    public static void main(String args[]){
        ReentrantLock lock=new ReentrantLock();

        new Thread(()->{
            lock.lock();
            System.out.println(Thread.currentThread().getName());
            lock.unlock();

        }).start();
        new Thread(()->{
            lock.lock();
            System.out.println(Thread.currentThread().getName());
            lock.unlock();
        }).start();
    }
}

